
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <signal.h>
#include <unistd.h>
#define _XOPEN_SOURCE 
#define __USE_XOPEN
#include <time.h>
#include <curl/curl.h>
#include <curl/easy.h>
#include <curl/multi.h>
#include "log.h"
#include "types.h"
#include "csvparser.h"
#include "http2hj212.h"
#include "yapiparser.h"
#include "enprodev.h"
#include "tcpclient.h"
#include "hj212packet.h"
#include "hj212parser.h"
#include "util.h"


char* socketbuff = "##0570QN=2024070514112115;ST=22;CN=2061;PW=123456;MN=abcdef123456789;Flag=7;PUNM=1;PON=1;CP=&&DataTime=20240705131020;a99054-Avg=0.0515;a21028-Avg=20.7146;a21001-Avg=4000.0000;a19002-Avg=0.0000;a99054-Avg=0.3879;a34004-Avg=51.1373;a01002-Avg=207.1000;a21028-Avg=9.6332;a01006-Avg=0.0000;a01007-Avg=5.9383;a34002-Avg=72.1511;a01007-Avg=1.8350;a01008-Avg=109.8000;a21001-Avg=617.4701;a21004-Avg=13.3379;a01002-Avg=84.2200;a01006-Avg=1020.6213;a01001-Avg=20.7000;a21005-Avg=0.6604;a05024-Avg=160.1802;a01001-Avg=24.8575;a21026-Avg=889.0229;a01008-Avg=207.0000;a19002-Avg=0.0000&&5440\r\n";
   

// static void* hj212_rcev(void* arg)
// {
//     printf("hj212_rcev thread start ...\n");
//     if ( arg )
//     {
//         TcpClient client = (TcpClient)arg;
//         Message* message = NULL;
//         unsigned int count = 0;

//         while ( 1 )
//         {
//             message = tcpclient_recv_msg(client);
//             if ( message )
//             {
//                 HJ212Msg* msg = (HJ212Msg*)message;
//                 ++count;
//                 printf("(%d) qn = %s\n", count, msg->qn);
//                 printf("(%d) ST = %d\n", count, msg->st);
//                 printf("(%d) CN = %d\n", count, msg->cn);
//                 printf("(%d) PW = %s\n", count, msg->pwd);
//                 printf("(%d) MN = %s\n", count, msg->mn);
//                 printf("(%d) Flag = %d\n", count, msg->flag);
//                 printf("(%d) Str = %s\n", count, msg->payload);
//                 HJ212_destroy_message(msg);
//             }

//             sleep(2);
//         }
//     }

//     printf("hj212_rcev thread end ...\n");
//     return NULL;
// }

static void* consumer_thread(void* arg)
{
    printf("consumer_thread start ...\n");

    if ( arg )
    {
        
        HtHJDev* dev = (HtHJDev*)arg;
        dev->hj212dev->init_ppmsg(dev->hj212dev->prdata, dev->hj212serconf->pwd, dev->hj212serconf->mn);
        
        while ( 1 )
        {
            dev->envprodev->condition(dev->envprodev);
            printf("envpd have data...\n");
            dev->hj212dev->from_envpro_socket(dev->hj212dev, dev->envprodev);
            dev->envprodev->destroy_list(dev->envprodev);
            //dev->hj212dev->print(dev->hj212dev);  // test
            //dev->hj212dev->destroy_list(dev->hj212dev);  // test

            HJ212PakNode* pnode = NULL;
            //printf("client start sending data...\n");
            for (dev->hj212dev->begin(dev->hj212dev); !dev->hj212dev->is_end(dev->hj212dev); dev->hj212dev->next(dev->hj212dev))
            {
                pnode = dev->hj212dev->current(dev->hj212dev);
                tcpclient_send_raw(dev->tcpclient, pnode->str, pnode->len);
                printf("send: %d --> %s\n", pnode->len, pnode->str);
            }
            // printf("client end send...\n");
            dev->hj212dev->destroy_list(dev->hj212dev);

            //sleep(2); // test
        }      
        
    }
    
    printf("consumer_thread end ...\n");
    return NULL;
}


#define PRODUCER_INTERVAL_SECONDS        20
#define MAIN_THREAD_INTERVAL_SEC         10
static void* producer_thread(void* arg)
{
    unsigned int interval = 0;
    printf("producer_thread start ...\n");

    if ( arg )
    {
        HtHJDev* dev = (HtHJDev*)arg;

        dev->envprodev->init(dev->envprodev, dev->yapiserconf, dev->yapiparser);
        csvparser_mutex_lock(dev->csvparser);
        dev->envprodev->request(dev->envprodev, dev->csvparser);
        //envprodev_simu_request(dev->envprodev);
        csvparser_mutex_unlock(dev->csvparser);
       
        while (1)
        {
            if ( interval >= PRODUCER_INTERVAL_SECONDS )
            {
                interval = 0;
                csvparser_mutex_lock(dev->csvparser);
                dev->envprodev->request(dev->envprodev, dev->csvparser);
                //envprodev_simu_request(dev->envprodev);
                csvparser_mutex_unlock(dev->csvparser);
                printf("production completion...\n");
            }
            
            interval += MAIN_THREAD_INTERVAL_SEC;
            sleep(MAIN_THREAD_INTERVAL_SEC);
        }

        
    }

    printf("producer_thread end ...\n");
    
    return NULL;
}






// static void* test_thread(void* arg)
// {
//     printf("test_thread start ...\n");
//     if ( arg )
//     {
//         TcpClient client = (TcpClient)arg;
//         // Message* message = NULL;
//         // int count = 0;
//         while ( 1 )
//         {
//             // message = tcpclient_recv_msg(client);
//             // if ( message )
//             // {
//             //     HJ212Msg* msg = (HJ212Msg*)message;
//             //     ++count;
//             //     printf("(%d) qn = %s\n", count, msg->qn);
//             //     printf("(%d) ST = %d\n", count, msg->st);
//             //     printf("(%d) CN = %d\n", count, msg->cn);
//             //     printf("(%d) PW = %s\n", count, msg->pwd);
//             //     printf("(%d) MN = %s\n", count, msg->mn);
//             //     printf("(%d) Flag = %d\n", count, msg->flag);
//             //     printf("(%d) Str = %s\n", count, msg->payload);
//             //     HJ212_destroy_message(msg);
//             // }

//             tcpclient_send_raw(client, socketbuff, strlen(socketbuff));
//             printf("client connected ok\n");
//             sleep(10);
//         }
//     }

//     return NULL;
// }

static char* testdatstr = "QN=2024070916025394;ST=22;CN=2061;PW=123456;MN=2111000140001;Flag=5;CP=&&DataTime=20240709150153;a21001-Avg=9.0155;a99054-Avg=0.0517;a21028-Avg=22.8628;a01007-Avg=7.3250;a01008-Avg=207.0000;a01001-Avg=20.7000&&";

#define QUERY_SCV_FILE_INTERVAL_SEC              18000
#define MAIN_THREAD_INTERVAL_SEC                 2
int main(void)
{
    unsigned int querycsvsec = 0;
    pthread_t conthread;
    pthread_attr_t conattr;
    pthread_t prothread;
    pthread_attr_t proattr;
    // pthread_t tcprecvthread;
    // pthread_attr_t tcprecvattr;
    Message* message = NULL;
    unsigned int count = 0;
    log_init();
    int ret = 0;


    ret = hj212_crc16(testdatstr, strlen(testdatstr));
    
    printf("%X   %X\n", ret >> 8,  ret & 0xFF);

    int dev_count = 0;
    HtHJDev* http2hjDev =  hthj_ctreat("config.ini");
    if ( http2hjDev )
    {
        CsvParser csvParser = csvparser_creat("device.csv");
        csvparser_read_csv_to_code_list(csvParser);
        printf("device has %d\n", csvparser_get_length(csvParser));
      

        YapiParser* yapiParser = yapiparser_create();
        EnvProDev* envproDev = envprodev_create();
        HJ212PackFact* hj212pf = hj212_pack_fact_new();
        HJ212Parser* hj21parser = hj212parser_new();
        TcpClient client = tcpclient_new(hj21parser);
        if ( yapiParser && envproDev && hj212pf && csvParser && client )
        {
            register_csv_parser(http2hjDev, csvParser);
            register_yapi_parser(http2hjDev, yapiParser);
            register_hj212_dev(http2hjDev, hj212pf);
            register_envpro_dev(http2hjDev, envproDev);
            register_tcp_client(http2hjDev, client);
            printf("register success...\n");
            if ( (ret = tcpclient_connect(client, http2hjDev->hj212serconf->ip, http2hjDev->hj212serconf->port)) == 0 )
            {
                printf("Connect failed! %d \n", ret);
                return -1;
            }

            pthread_attr_init(&conattr);
            pthread_attr_setdetachstate(&conattr, PTHREAD_CREATE_DETACHED);
            pthread_create(&conthread, &conattr, consumer_thread, (void*)http2hjDev);
            pthread_attr_init(&proattr);
            pthread_attr_setdetachstate(&proattr, PTHREAD_CREATE_DETACHED);
            pthread_create(&prothread, &proattr, producer_thread, (void*)http2hjDev);
            pthread_attr_destroy(&conattr);
            pthread_attr_destroy(&proattr);

            // pthread_attr_init(&tcprecvattr);
            // pthread_attr_setdetachstate(&tcprecvattr, PTHREAD_CREATE_DETACHED); 
            // pthread_create(&tcprecvthread, &tcprecvattr, test_thread, client);
            while ( 1 )
            {
                printf("Thread\n");
                message = tcpclient_recv_msg(client);
                printf("received message end\n");
                if ( message )
                {
                    HJ212Msg* msg = (HJ212Msg*)message;
                    ++count;
                    printf("(%d) qn = %s\n", count, msg->qn);
                    printf("(%d) ST = %d\n", count, msg->st);
                    printf("(%d) CN = %d\n", count, msg->cn);
                    printf("(%d) PW = %s\n", count, msg->pwd);
                    printf("(%d) MN = %s\n", count, msg->mn);
                    printf("(%d) Flag = %d\n", count, msg->flag);
                    printf("(%d) Str = %s\n", count, msg->payload);
                    HJ212_destroy_message(msg);
                }   

                // querycsvsec += MAIN_THREAD_INTERVAL_SEC;
                // if ( querycsvsec >= QUERY_SCV_FILE_INTERVAL_SEC )
                // {
                //     csvparser_mutex_lock(csvParser);
                //     csvparser_read_csv_to_code_list(csvParser);
                //     csvparser_mutex_unlock(csvParser);
                //     querycsvsec = 0;
                //     LOG_I("update csv file...\n");
                // }

                sleep(MAIN_THREAD_INTERVAL_SEC);
            }
                
        }
        else
           http2hjDev = hthj_destroy(http2hjDev);
        
        
    }
    
    printf(" is end\n\r");
    return 0;
}